package com.gallop.akka;

import akka.actor.ActorSystem;
import akka.cluster.Cluster;
import com.gallop.akka.config.ActorInitializer;
import com.gallop.akka.config.AkkaPropInterceptors;
import com.gallop.akka.serialize.ActorSerializableMessage;
import com.gallop.akka.support.DefaultActorContextSupport;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.flyway.FlywayAutoConfiguration;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.ComponentScans;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Akka autoConfiguration for Spring
 *
 * @author pipi
 */
@Configuration
@EnableConfigurationProperties(AkkaProperties.class)
@AutoConfigureAfter(FlywayAutoConfiguration.class)
@ComponentScans({
        @ComponentScan("com.gallop.akka"),
})
public class AkkaAutoConfiguration implements ApplicationContextAware {
    public static final String CLUSTER_DEFAULT_NAME = "akka-default-cluster-system";

    private ApplicationContext applicationContext;
    @Autowired
    private AkkaProperties akkaProperties;
    @Autowired
    private ActorInitializer actorInitializer;
    @Autowired
    private ServerProperties serverProperties;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @PostConstruct
    public void init() throws Exception {
        //register actors on cluster
        //applicationContext.getBean()

        // reassign property values in code.
        propInterceptors().apply(akkaProperties);
        // register actors when start up
        actorInitializer.register(actorContextSupport(), actorSystem(), akkaProperties);
    }

    @Bean
    public ActorContextSupport actorContextSupport() {
        return new DefaultActorContextSupport();
    }

    @Bean
    @ConditionalOnMissingBean(AkkaPropInterceptors.class)
    public AkkaPropInterceptors propInterceptors() {
        return new AkkaPropInterceptors();
    }

    @Bean
    @ConditionalOnMissingBean(ActorInitializer.class)
    public ActorInitializer actorInitializer() {
        return (contextSupport, system, properties) -> {
        };
    }

    @Bean(destroyMethod = "terminate")
    public ActorSystem actorSystem() throws Exception {
        String actorSystemName = akkaProperties.getSystemName();
        if (!StringUtils.hasLength(actorSystemName)) {
            actorSystemName = CLUSTER_DEFAULT_NAME;
        }
        Resource conf = akkaProperties.getConf();
        ActorSystem system;
        if (conf == null) {
            system = ActorSystem.create(actorSystemName);
        } else {
            system = ActorSystem.create(actorSystemName, combinedConfig());
        }
        SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(system).initialize(applicationContext);
        if (ActorProviderType.cluster.equals(akkaProperties.getActorProvider())
                || ActorProviderType.cluster_sharding.equals(akkaProperties.getActorProvider())) {
            return Cluster.get(system).system();
        }
        return system;
    }

//    @Bean
//    public Cluster cluster() throws Exception {
//        return Cluster.get(actorSystem());
//    }

    /**
     * combine config from config file & settings
     *
     * @return AKKA config
     * @throws Exception exception
     */
    private Config combinedConfig() throws Exception {
        List<String> configItems = new ArrayList<>();
        if (akkaProperties.getActorProvider() != null) {
            configItems.add("akka.actor.provider=\"" + akkaProperties.getActorProvider().getValue() + "\"");
            if (ActorProviderType.local != akkaProperties.getActorProvider()) {
                // 需要 pom 额外引入netty
//                configItems.add("akka.remote.netty.tcp.port=" + akkaProperties.getListenPort());
                configItems.add("akka.remote.artery.canonical.hostname=" + akkaProperties.getListenHost());
//                configItems.add("akka.remote.artery.canonical.port=" + akkaProperties.getListenPort());
                configItems.add("akka.remote.artery.canonical.port=" +
                        akkaProperties.getPortCombineMethod().getCombiner().combine(serverProperties.getPort(), akkaProperties.getListenPort())
                );

            }
        }
        Long passivateSeconds = akkaProperties.getPassivateIdleEntityAfter();
        configItems.add("akka.cluster.sharding.passivate-idle-entity-after=" +
                (passivateSeconds <= 0 ? "off" : passivateSeconds + "s"));
        if (akkaProperties.getEnableSerializer()) {
            // 注册serializer
            configItems.add("akka.actor.serializers." +
                    akkaProperties.getSerializerType().getSerializerName() + "="
                    + akkaProperties.getSerializerType().getSerializerType());
            // 注册binding
            configItems.add("akka.actor.serialization-bindings.\"" + ActorSerializableMessage.class.getName() + "\"=" +
                    akkaProperties.getSerializerType().getSerializerName());
            // 注册自定义消息序列化
            if (!CollectionUtils.isEmpty(akkaProperties.getSerializeExtraMessages())) {
                akkaProperties.getSerializeExtraMessages().forEach(type -> {
                    // 扫描package
                    configItems.add("akka.actor.serialization-bindings.\"" + type.getName() + "\"=" +
                            akkaProperties.getSerializerType().getSerializerName());
                });
            }
        } else {
            configItems.add("akka.actor.allow-java-serialization=\"" + "on" + "\"");
        }
        if (StringUtils.hasLength(akkaProperties.getClusterRole())) {
            configItems.add("akka.cluster.roles=" + "[\"" + akkaProperties.getClusterRole() + "\"]");
        }
        if (akkaProperties.getClusterSeedNodes() != null && !akkaProperties.getClusterSeedNodes().isEmpty()) {
            configItems.add("akka.cluster.seed-nodes=" + akkaProperties.getClusterSeedNodes().stream().map(s -> "\"" + s + "\"").collect(Collectors.joining(",", "[", "]")));
        }
        if (akkaProperties.getExtensions() != null && !akkaProperties.getExtensions().isEmpty()) {
            configItems.add("akka.extensions=" + akkaProperties.getExtensions().stream().map(s -> "\"" + s + "\"").collect(Collectors.joining(",", "[", "]")));
        }
        if (akkaProperties.getHeartbeatPause() > 0) {
            configItems.add("akka.cluster.failure-detector.acceptable-heartbeat-pause=" + akkaProperties.getHeartbeatPause() + "s");
        }
        Config appConfig = ConfigFactory.parseString(String.join("\r\n", configItems));
        if (akkaProperties.getConf() != null) {
            Config regularConfig = ConfigFactory.parseURL(akkaProperties.getConf().getURL());
            appConfig = appConfig.withFallback(regularConfig);
        }
        return appConfig;
    }
}
